package spring.cloud.service.task.service.impl;

import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import spring.cloud.service.task.domain.executor.ExecutorInfo;
import spring.cloud.service.task.domain.executor.ExecutorRunningInfo;
import spring.cloud.service.task.domain.executor.ExecutorTaskItemAssignInfo;
import spring.cloud.service.task.domain.task_executor.TaskExecutorStatus;
import spring.cloud.service.task.dto.TaskDto;
import spring.cloud.service.task.service.ExecutorService;
import spring.cloud.service.task.service.TaskService;
import spring.cloud.service.task.service.ZooKeeperService;
import spring.cloud.tasks.common.Symbols;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.ExecutorNodePaths;
import spring.cloud.tasks.common.nodepaths.TaskNodePaths;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Service
public class ExecutorServiceImpl implements ExecutorService {
    @Resource
    private ZooKeeperService zooKeeperService;
    @Resource
    private TaskService taskService;


    @Override
    public void removeExecutor(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        curator.delete(ExecutorNodePaths.getExecutorSubNodePath(executorId));
        List<String> taskIdList = taskService.getTaskIdListFromZooKeeper(nameSpace);
        if (CollectionUtils.isEmpty(taskIdList)) {
            return;
        }
        for (String taskId : taskIdList) {
            String executorNode = TaskNodePaths.getTaskExecutorNodePath(taskId, executorId);
            curator.delete(executorNode);
        }
    }


    @Override
    public void extractTraffic(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        checkExists(curator, executorId);
        curator.merge(ExecutorNodePaths.getExecutorOwnTrafficNodePath(executorId), ExecutorInfo.own_traffic_n);
    }

    @Override
    public void recoverTraffic(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        checkExists(curator, executorId);
        curator.merge(ExecutorNodePaths.getExecutorOwnTrafficNodePath(executorId), ExecutorInfo.own_traffic_y);
    }

    @Override
    public void restart(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        checkExists(curator, executorId);
        String restartNodePath = ExecutorNodePaths.getExecutorRestartNodePath(executorId);
        curator.delete(restartNodePath);
        curator.create(restartNodePath);
    }

    private void checkExists(Curator curator, String executorId) {
        if (!curator.checkExists(ExecutorNodePaths.getExecutorSubNodePath(executorId))) {
            throw new RuntimeException("The executor(" + executorId + ") is not existed.");
        }
    }

    @Override
    public void shardAll(String nameSpace) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        String shardAllAtOnceNodePath = "/shardAllAtOnce";
        curator.delete(shardAllAtOnceNodePath);
        curator.create(shardAllAtOnceNodePath);
    }

    @Override
    public List<ExecutorInfo> getExecutorInfoList(String nameSpace) throws RuntimeException {
        return getExecutorInfoList(nameSpace, null);
    }

    @Override
    public List<ExecutorInfo> getExecutorInfoList(String nameSpace, String status) {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        List<String> executorIdList = curator.getChildren(ExecutorNodePaths.getRoot());
        if (executorIdList == null || executorIdList.isEmpty()) {
            return new ArrayList<ExecutorInfo>();
        }
        List<ExecutorInfo> executorInfoList = new ArrayList<ExecutorInfo>();
        for (String executorId : executorIdList) {
            ExecutorInfo executorInfo = getExecutorInfo(curator, executorId);
            if (status == null || executorInfo.getStatus() == status) {
                executorInfoList.add(executorInfo);
            }
        }

        return executorInfoList;
    }

    @Override
    public ExecutorInfo getExecutorInfo(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        if (!curator.checkExists(ExecutorNodePaths.getExecutorSubNodePath(executorId))) {
            return null;
        }
        return getExecutorInfo(curator, executorId);
    }

    private ExecutorInfo getExecutorInfo(Curator curator, String executorId) {
        ExecutorInfo executorInfo = new ExecutorInfo(executorId);
        // lastBeginTime
        String lastBeginTime = curator.getData(ExecutorNodePaths.getExecutorSubNodePath(executorInfo.getExecutorId(), "lastBeginTime"));
        executorInfo.setLastBeginTime(lastBeginTime);
        //
        String ip = curator.getData(ExecutorNodePaths.getExecutorIpNodePath(executorId));
        executorInfo.setIp(ip);
        if (StringUtils.isNotBlank(ip)) {
            executorInfo.setStatus(TaskExecutorStatus.ONLINE);
        } else {
            executorInfo.setStatus(TaskExecutorStatus.OFFLINE);
        }
        // 是否有流量
        String ownTraffic = curator.getData(ExecutorNodePaths.getExecutorOwnTrafficNodePath(executorId));
        executorInfo.setOwnTraffic(ExecutorInfo.own_traffic_y.equals(ownTraffic));
        // 如果restart结点存在，restart结点存在的时间<300s，executor状态列显示RESTARTING；
        long now = System.currentTimeMillis();
        String executorRestartNodePath = ExecutorNodePaths.getExecutorRestartNodePath(executorId);
        long restartTriggerTime = curator.getCtime(executorRestartNodePath);
        long maxRestartInv = 1000 * 30;
        executorInfo.setRestarting(0 != restartTriggerTime && now - restartTriggerTime < maxRestartInv);
        return executorInfo;
    }

    @Override
    public ExecutorTaskItemAssignInfo getExecutorTaskItemAssignInfo(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        ExecutorTaskItemAssignInfo executorTaskItemAssignInfo = new ExecutorTaskItemAssignInfo(executorId);
        List<TaskDto> taskDtoList = taskService.getTaskDtoList(nameSpace);
        for (TaskDto taskDto : taskDtoList) {
            String taskId = taskDto.getTaskId();
            String taskExecutorNodePath = TaskNodePaths.getTaskExecutorsNodePath(taskId);
            if (!curator.checkExists(taskExecutorNodePath)) {
                continue;
            }
            String sharding = curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, "sharding"));
            if (StringUtils.isNotBlank(sharding)) {
                // concat executorSharding
                executorTaskItemAssignInfo.getTaskIdToTaskItemIndexListMap().put(taskId, sharding);
                // calculate totalLoad
                String loadLevelNode = curator.getData(TaskNodePaths.getTaskBaseInfoNodePath(taskId, "loadLevel"));
                Integer loadLevel = Integer.valueOf(loadLevelNode);
                int shardingItemNum = sharding.split(Symbols.COMMA).length;
                int taskLoad = shardingItemNum * loadLevel;
                int totalLoadLevel = executorTaskItemAssignInfo.getTotalLoadLevel();
                executorTaskItemAssignInfo.setTotalLoadLevel(totalLoadLevel + taskLoad);
            }
        }

        return executorTaskItemAssignInfo;
    }

    @Override
    public ExecutorRunningInfo getExecutorRunningInfo(String nameSpace, String executorId) throws RuntimeException {
        Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
        ExecutorRunningInfo executorRunningInfo = new ExecutorRunningInfo();
        executorRunningInfo.setExecutorId(executorId);
        List<TaskDto> taskDtoList = taskService.getTaskDtoList(nameSpace);
        for (TaskDto taskDto : taskDtoList) {
            String taskId = taskDto.getTaskId();
            String taskExecutorNodePath = TaskNodePaths.getTaskExecutorsNodePath(taskId);
            if (!curator.checkExists(taskExecutorNodePath)) {
                continue;
            }
            String sharding = curator.getData(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId, "sharding"));
            if (StringUtils.isEmpty(sharding)) {
                continue;
            }
            Set<String> shardingItems = Arrays.stream(StringUtils.split(sharding, Symbols.COMMA)).distinct().collect(Collectors.toSet());
            if (!shardingItems.isEmpty()) {
                getTaskIdRunningTaskItemIdList(curator, taskDto, shardingItems, executorRunningInfo);
            }
        }
        return executorRunningInfo;
    }

    private void getTaskIdRunningTaskItemIdList(Curator curator, TaskDto taskDto, Set<String> shardingItems, ExecutorRunningInfo executorRunningInfo) {
        // 对于不上报运行状态的任务，所有分配的分片均作为潜在运行中的任务
        List<String> runningTaskItemIdList = new ArrayList<>();
        for (String item : shardingItems) {
            //if (curator.checkExists(TaskNode.getTaskItemNodePath(taskDto.getTaskId(), item, "running")))
            //判断当前的任务分片是否存在正在跑的任务分片
            {
                runningTaskItemIdList.add(item);
            }
        }
        if (!runningTaskItemIdList.isEmpty()) {
            executorRunningInfo.getTaskIdToRunningTaskItemIdListMap().put(taskDto.getTaskId(), StringUtils.join(runningTaskItemIdList, ','));
        }
    }


}
